// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include <fmt/format.h>

#include "coin2/exchange/base/feed/live_parallel/system.h"

#include "coin2/exchange/base/feed/replay/distributor.h"
#include "coin2/exchange/base/feed/ws/subscriber_manager.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/flow/symbol_group.h"

namespace coin2::exchange::base::feed {

using coin2::exchange::base::feed::replay::FeedRelayDistributor;
using coin2::exchange::base::feed::replay::FeedRelayTelemetryDistributor;
using coin2::exchange::base::feed::counter::FeedCounter;

LiveParallelFeedSubsystem::LiveParallelFeedSubsystem(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    const coin2::exchange::di::ConstantInjection& ci,
    const coin2::exchange::feed::FeedSystemConfig& config,
    coin2::exchange::base::executor::LiveAsyncExecutor* live_executor,
    IFeedSubscriber* callback,
    FeedCounter* counter)
    : BaseLiveFeedSubsystem(system_name, mea, live_executor),
      num_workers_(config.websocket().num_workers()),
      feed_arbitration_(config.websocket().arbitration()),
      ci_(ci),
      config_(config),
      callback_(callback),
      feed_counter_(counter) {
  subscriber_ = std::make_shared<SubscriberManager>(
      ci.GetFeedSubscriptionSpec(mea),
      config.websocket(),
      coin2::flow::TopicType::CONNECTION_TOPIC,
      live_executor,
      config.raw_parser().enable_telemetry());

  auto temp_products = ProductRequestsFromConfig(
      mea_.String(),
      config.products(),
      GetCurrentTimestamp(),
      false);  // die_on_failure
  if (config.products().has_norm_format()
      && config.products().norm_format().find("*") != std::string::npos) {
    products_ = FilterProductWithInSymbolGroups(mea_, temp_products);
  } else {
    products_ = std::move(temp_products);
  }

  for (auto& prod : products_) {
    auto symbol = prod->feed_native_symbol();
    char group = coin2::flow::SymbolGroupConfig::GetInstance().GetGroupForRelativeNorm(
        mea_,
        prod->relative_norm());
    subscriber_->RequestNativeSymbol(symbol, group);
  }
}

void LiveParallelFeedSubsystem::StartAsync() {
  subscriber_->CreateSubscribers();
  const auto topics = subscriber_->GetTopics();
  live_executor_->injector().RegisterIoc(topics);

  for (const auto& topic : topics) {
    executor::TopicRecordSubscriber* topic_record_subscriber = nullptr; 
    auto t = coin2::flow::TopicStringConfig::StripWokerFromConnectionTopic(topic);
    auto iter = topic2subscriber_.find(t);
    if (iter == topic2subscriber_.end()) {
      topic_record_subscriber = this->NewTopicRecordSubscriber();
      topic2subscriber_[t] = topic_record_subscriber;
    } else {
      topic_record_subscriber = iter->second;
    }

    topic_register_.RegisterTopic(topic, topic_record_subscriber, live_executor_->distributor());
  }

  // check topic
  for (const auto& topic : topics) {
    subscriber_->RegisterPubTopic(topic);
  }
  subscriber_->StartAsync();
  health_.status = HealthStatus::HEALTH_STARTING;
  health_.since = GetCurrentTimestamp();
}

void LiveParallelFeedSubsystem::StopAsync() { subscriber_->StopAsync(); }

std::vector<IProduct*> LiveParallelFeedSubsystem::GetProducts() const {
  std::vector<IProduct*> out;
  for (const auto& p : products_) {
    out.push_back(p.get());
  }
  return out;
}

void LiveParallelFeedSubsystem::GetHealthReport(int indent, std::string* report) const {
  // [HealthReport]
  //     [Feed]
  //         - Huobi (OK)
  //         - Okex (STARTING, uptime=23h)
  //               a1 (..)
  HealthReport health{};
  subscriber_->GetHealthReport(indent + 1, &health);

  report->append(std::string(indent * 4, ' '));
  std::string info = fmt::format(
      "- {system} ({health}, uptime='{uptime}', alive_connection={connection_num}, symbols={symbol_num})\n",
      fmt::arg("system", system_name()),
      fmt::arg("health", HealthStatusName(this->health_.status)),
      fmt::arg("uptime", DebugStringFromDuration(GetCurrentTimestamp() - health_.since)),
      fmt::arg("connection_num", health.connection_num),
      fmt::arg("symbol_num", health.symbol_num));
  report->append(info).append(health.report);
}

bool LiveParallelFeedSubsystem::BanProduct(const IProduct* product) {
  std::vector<const IProduct*> products{product};
  return BanProduct(products);
}

bool LiveParallelFeedSubsystem::BanProduct(const std::vector<const IProduct*>& products) {
  std::vector<const IProduct*> checked_products;
  for (auto& product : products) {
    if (product->market() == mea_.market &&
        product->exchange() == mea_.exchange) {
      checked_products.push_back(product);
    }
  }

  if (checked_products.empty()) return false;

  for (auto& processor : processors_) {
    processor->RemoveProduct(checked_products);
  }
  subscriber_->BanProduct(checked_products);

  return true;
}

bool LiveParallelFeedSubsystem::BanProduct(const std::vector<IProduct*>& products) {
  std::vector<const IProduct*> const_products;
  const_products.insert(const_products.end(), products.begin(), products.end());
  return BanProduct(const_products);
}

executor::TopicRecordSubscriber* LiveParallelFeedSubsystem::NewTopicRecordSubscriber() {
  executor::TopicRecordSubscriber* subscriber = nullptr;
  auto processor = factory::feed::NewRawProcessor(
      ci_, system_name_, mea_, config_.raw_parser(), config_.recipe());
  int book_level = 5;  // default value
  if (config_.raw_parser().has_book_level()) {
    book_level = config_.raw_parser().book_level();
  }
  std::unique_ptr<FeedDistributor> distributor;
  if (config_.raw_parser().enable_telemetry()) {
    distributor = std::make_unique<FeedRelayTelemetryDistributor>(
        system_name_,
        mea_,
        live_executor_->MainIoc(),
        book_level,
        feed_counter_);
  } else {
    distributor = std::make_unique<FeedRelayDistributor>(
        system_name_,
        mea_,
        live_executor_->MainIoc(),
        book_level,
        feed_counter_);
  }
  processor->mutable_upd_mgr()->SetDistibutor(std::move(distributor));
  for (const auto& prod : products_) {
    processor->AddProduct(*prod, callback_);
  }

  subscriber = processor.get();
  processors_.emplace_back(std::move(processor));

  if (feed_arbitration_) {
    auto arbitrator = std::make_unique<executor::TopicRecordArbitrator>(
        subscriber, ci_.GetArbitrationConfigs().GetWindowSize(mea_));
    subscriber = arbitrator.get();
    arbitrators_.emplace_back(std::move(arbitrator));
  }

  return subscriber;
}

}  // namespace coin2::exchange::base::feed
